package chord;

/**
 *
 * @author prabhu
 */
import java.io.File;
import java.io.FileNotFoundException;
import java.util.Enumeration;
import java.util.Scanner;
import java.util.Vector;

import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.ChannelClosedException;
import org.jgroups.ChannelNotConnectedException;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;

class Node extends ReceiverAdapter
{
    int intID;
    int intGroupSize;

    FingerTable fingerTable;
    Vector<Integer> keys;
    NodeAddressMap nodeMap;

    JChannel channel;
    Vector<Address> view;

    GUI gui;

    //long lngSearchNanoTime;
    //long lngSearchResultNanoTime;

    public Node(int intNodeID)
    {
	intID = intNodeID;
	nodeMap = new NodeAddressMap();
	fingerTable = new FingerTable();
	keys = new Vector<Integer>();
    }

    public void initiate(String strFileName) throws Exception
    {
	Logger.log(LogType.DEBUG, "Starting new node " + intID);

	channel = new JChannel();
	channel.setOpt(Channel.LOCAL, Boolean.FALSE);
	channel.setReceiver(this);
	channel.setName(Integer.toString(intID));
	channel.connect("MyChord");
	channel.getState(null, 10000);

	intGroupSize = channel.getView().getMembers().size();

	if (intGroupSize == 1)
	{
	    Logger.log(LogType.DEBUG, "DEBUG : I am the first node");
	    readKeysFromFile(strFileName);
	    bootstrap();
	    nodeMap.addrNodes[intID] = channel.getAddress();
	}
	else
	{
	    Logger.log(LogType.DEBUG, "DEBUG : Groups size = " + intGroupSize);
	    bootstrap();

	    ChordMessage bootstrapMsg = new ChordMessage(intID,
		    MessageType.BOOTSTRAP, null);
	    sendMessage(null, null, bootstrapMsg);
	}

	eventLoop();
	channel.close();
    }

    public void viewAccepted(View newView)
    {
	if (view != null)
	{
	    if (view.size() > newView.size())
		handleNodeFailure(newView);
	    else
		Logger.log(LogType.DEBUG, "DEBUG : New Node connected");
	}
	view = newView.getMembers();
    }

    public void receive(Message msg)
    {
	ChordMessage chordMsg = (ChordMessage) msg.getObject();

	Logger.log(LogType.MESSAGE_RECEIVED, chordMsg.intNodeID + " "
		+ chordMsg.msgType);

	switch (chordMsg.msgType)
	    {
	    case BOOTSTRAP:
	    {
		// If I am successor for chord.intNodeID then send finger table
		// and
		// update own finger table
		int intSuccessor = fingerTable
			.findSuccessor(chordMsg.intNodeID);

		if (intSuccessor == intID)
		{
		    sendBootStrapResponse(msg.getSrc());

		    sendKeys(chordMsg.intNodeID, msg.getSrc());
		}
	    }
		break;

	    case BOOTSTRAP_RESPONSE:
	    {
		nodeMap = (NodeAddressMap) chordMsg.object;
		nodeMap.addrNodes[intID] = channel.getAddress();

		fingerTable.updateOwnFingerTable();

		fingerTable.show();

		sendStatusUp();
	    }
		break;

	    case STATUS_UP:
	    {
		nodeMap.addrNodes[chordMsg.intNodeID] = msg.getSrc();

		fingerTable.updateOwnFingerTable();
		fingerTable.show();
	    }
		break;

	    case KEYS:
	    {
		KeysVector keysVector = (KeysVector) chordMsg.object;
		this.keys = keysVector.keys;

		Enumeration<Integer> e = keys.elements();

		while (e.hasMoreElements())
		{
		    Logger.log(LogType.RAW, "The elements are: "
			    + e.nextElement());
		}
	    }
		break;

	    case EXIT_TRANSFER_KEYS:
	    {
		KeysVector keysVector = (KeysVector) chordMsg.object;
		keys.addAll(keysVector.keys);
	    }
		break;

	    case EXIT:
	    {
		nodeMap.addrNodes[chordMsg.intNodeID] = null;

		fingerTable.updateOwnFingerTable();
		fingerTable.show();
	    }
		break;

	    case SEARCH_RESULT:
	    {
		//lngSearchResultNanoTime = System.nanoTime();
		SearchMessage msgSearch = (SearchMessage) chordMsg.object;
		msgSearch.stopTimer();
		msgSearch.intHopCount++;
		HandleSearchResult(chordMsg.intNodeID, msgSearch);
	    }
		break;

	    case SEARCH_QUERY:
	    {
		SearchMessage msgSearch = (SearchMessage) chordMsg.object;
		msgSearch.intHopCount++;

		if (keys.contains(new Integer(msgSearch.intKeyToSearch)))
		{
		    if (msgSearch.addrSrc.equals(channel.getAddress()))
	            {
			msgSearch.lngNanoTime = System.nanoTime() - msgSearch.lngNanoTime;
	                gui.displaySearchResult(intID, msgSearch);	                
	            }
	            else
	            {
	                SendSearchResultBackToOriginalNode(msgSearch);
	            }
		}
		else
		{
		    SendSearchFailedBackToOriginalNode(msgSearch);
		}
	    }
		break;

	    case SEARCH_QUERY_FORWARD:
	    {
		SearchMessage msgSearch = (SearchMessage) chordMsg.object;
		msgSearch.intHopCount++;
		searchKey(msgSearch, true);
	    }
		break;

	    case SEARCH_FAILED:
	    {
		//lngSearchResultNanoTime = System.nanoTime();
		gui.displaySearchResult(-1);
	    }
		break;

	    default:
	    }

    }

    private void HandleSearchResult(int intNodeID, SearchMessage msgSearch)
    {
	gui.displaySearchResult(intNodeID, msgSearch);
    }

    public KeysVector getKeys(int intNodeID)
    {
	KeysVector keysVector = new KeysVector();
	int intPredeccessor = fingerTable.findPredeccessor(intNodeID);

	Logger.log(LogType.RAW, "intPredeccessor" + intPredeccessor);

	Enumeration<Integer> e = keys.elements();

	while (e.hasMoreElements())
	{
	    Integer key = e.nextElement();
	    int keyHashValue = getHash(key.intValue());

	    Logger.log(LogType.RAW, " isBetween("
		    + intPredeccessor
		    + "  "
		    + (intNodeID)
		    + "  "
		    + keyHashValue
		    + " "
		    + fingerTable.isBetween(intPredeccessor, intNodeID,
			    keyHashValue));

	    if (fingerTable.isBetween(intPredeccessor, intNodeID, keyHashValue))
	    {
		keysVector.keys.add(key);
	    }
	}

	/*
	 * e = keysVector.keys.elements();
	 * 
	 * while (e.hasMoreElements()) { keys.remove(e.nextElement()); }
	 */

	keys.removeAll(keysVector.keys);

	e = keys.elements();

	while (e.hasMoreElements())
	{
	    Logger.log(LogType.RAW, "The elements are: " + e.nextElement());
	}

	return keysVector;
    }

    private int getHash(int intKey)
    {
	return (intKey % FingerTable.intChordSize);
    }

    private void sendKeys(int intNodeID, Address dstAddress)
    {
	KeysVector keys = getKeys(intNodeID);

	ChordMessage chordMsg = new ChordMessage(intID, MessageType.KEYS, keys);

	sendMessage(dstAddress, null, chordMsg);
    }

    public void sendStatusUp()
    {
	ChordMessage chordMsg = new ChordMessage(intID, MessageType.STATUS_UP,
		null);

	sendMessage(null, null, chordMsg);
    }

    public void sendBootStrapResponse(Address dstAddress)
    {
	ChordMessage chordMsg = new ChordMessage(intID,
		MessageType.BOOTSTRAP_RESPONSE, nodeMap);

	sendMessage(dstAddress, null, chordMsg);
    }

    public void sendExitMessage()
    {
	int intSuccessor = fingerTable.findSuccessor(intID);
	Address addrSuccessor = nodeMap.addrNodes[intSuccessor];
	KeysVector keyVector = new KeysVector();
	keyVector.keys = this.keys;

	ChordMessage msgExitTransferKeys = new ChordMessage(intID,
		MessageType.EXIT_TRANSFER_KEYS, keyVector);
	ChordMessage msgExit = new ChordMessage(intID, MessageType.EXIT,
		keyVector);

	sendMessage(addrSuccessor, null, msgExitTransferKeys);

	sendMessage(null, null, msgExit);

	channel.close();
	System.exit(0);
    }

    public int getNodeID(Address addr)
    {
	int intNodeID = -1;

	if (addr != null)
	    for (int i = 0; i < FingerTable.intChordSize; i++)
	    {
		if (nodeMap.addrNodes[i] != null)
		    if (nodeMap.addrNodes[i].compareTo(addr) == 0)
		    {
			intNodeID = i;
			break;
		    }
	    }
	return intNodeID;
    }

    public void handleNodeFailure(View newView)
    {
	Address addrNodeLeft = null;

	Logger.log(LogType.RAW, view.size() + " View.size() " + newView.size()
		+ "  newView.size()");

	for (int i = 0; i < view.size(); i++)
	{
	    if (!newView.containsMember(view.get(i)))
	    {
		addrNodeLeft = view.get(i);
		Logger.log(LogType.DEBUG, "Node Left " + addrNodeLeft);
		break;
	    }
	}

	int intNodeFailed = getNodeID(addrNodeLeft);

	if (intNodeFailed > 0)
	{
	    nodeMap.addrNodes[intNodeFailed] = null;

	    fingerTable.updateOwnFingerTable();
	    fingerTable.show();
	}
    }

    private void eventLoop()
    {
	while (true)
	{
	    try
	    {

	    }
	    catch (Exception e)
	    {
	    }
	}

    }

    public void bootstrap()
    {
	fingerTable.initFingerTable(this);
	fingerTable.show();
    }

    public void readKeysFromFile(String strFileName)
    {
	Scanner fileScan;
	try
	{
	    fileScan = new Scanner(new File(strFileName));

	    int intKey;

	    while (fileScan.hasNext())
	    {
		intKey = fileScan.nextInt();
		keys.add(new Integer(intKey));
	    }
	}
	catch (FileNotFoundException e)
	{
	    e.printStackTrace();
	}

	Enumeration<Integer> e = keys.elements();

	while (e.hasMoreElements())
	{
	    System.out.println("The elements are: " + e.nextElement());
	}

    }

    public int searchKey(SearchMessage msgSearch, boolean bForward)
    {
	int intNodeID = -1;
	int intKeyHash;
	// boolean bForward;

	intKeyHash = getHash(msgSearch.intKeyToSearch);
	int intNextSuccessor = fingerTable.findNextSuccessor(intKeyHash);

	Logger.log(LogType.RAW, " NextSuccessor " + intNextSuccessor);

	if (intKeyHash == intID || intNextSuccessor == intID)
	{
	    if (msgSearch.addrSrc.equals(channel.getAddress()))
	    {
		//lngSearchResultNanoTime = System.nanoTime();
		msgSearch.stopTimer();
		if (keys.contains(new Integer(msgSearch.intKeyToSearch)))
		    gui.displaySearchResult(intID);
		else
		    gui.displaySearchResult(-1);
	    }
	    else
	    {
		SendSearchResultBackToOriginalNode(msgSearch);
	    }

	}
	else if (bForward)
	{
	    boolean bSearchConfident = fingerTable
		    .isKeyLessThanNextSuccessor(intKeyHash);

	    ForwardSearchQuery(msgSearch, nodeMap.addrNodes[intNextSuccessor],
		    !bSearchConfident);
	}
	else
	{
	    SendSearchFailedBackToOriginalNode(msgSearch);
	}

	return intNodeID;
    }

    private void ForwardSearchQuery(SearchMessage msgSearch,
	    Address addrNextSuccessor, boolean bForward)
    {
	MessageType msgType;
	if (bForward)
	    msgType = MessageType.SEARCH_QUERY_FORWARD;
	else
	    msgType = MessageType.SEARCH_QUERY;

	ChordMessage msg = new ChordMessage(intID, msgType, msgSearch);

	sendMessage(addrNextSuccessor, null, msg);

	Logger.log(LogType.DEBUG, "Forwarded query to search : "
		+ msgSearch.intKeyToSearch + " From "
		+ getNodeID(msgSearch.addrSrc) + " to next node : "
		+ getNodeID(addrNextSuccessor));

    }

    private void SendSearchResultBackToOriginalNode(SearchMessage msgSearch)
    {
	ChordMessage chordMsg = new ChordMessage(intID,
		MessageType.SEARCH_RESULT, msgSearch);

	sendMessage(msgSearch.addrSrc, null, chordMsg);

	Logger.log(LogType.DEBUG, "Found Key  : " + msgSearch.intKeyToSearch
		+ " Sent Back to " + getNodeID(msgSearch.addrSrc));

    }

    private void SendSearchFailedBackToOriginalNode(SearchMessage msgSearch)
    {
	ChordMessage chordMsg = new ChordMessage(intID,
		MessageType.SEARCH_FAILED, msgSearch);

	sendMessage(msgSearch.addrSrc, null, chordMsg);

	Logger.log(LogType.DEBUG, "Search failed for Key  : "
		+ msgSearch.intKeyToSearch + " Sent Back to "
		+ getNodeID(msgSearch.addrSrc));

    }

    public void setGUICallback(GUI gui)
    {
	this.gui = gui;
    }

    public void search(int intKeyToSearch, boolean bForward)
    {
	SearchMessage msgSearch = new SearchMessage();
	msgSearch.intKeyToSearch = intKeyToSearch;
	msgSearch.addrSrc = channel.getAddress();

	msgSearch.lngNanoTime = System.nanoTime();
	searchKey(msgSearch, bForward);
    }

    public void sendMessage(Address dst, Address src, ChordMessage chordMessage)
    {
	try
	{
	    channel.send(dst, src, chordMessage);
	}
	catch (ChannelNotConnectedException e)
	{
	    Logger.log(LogType.DEBUG, "Failed to send Message "
		    + chordMessage.msgType + " to " + chordMessage.intNodeID);
	    e.printStackTrace();
	}
	catch (ChannelClosedException e)
	{
	    Logger.log(LogType.DEBUG, "Failed to send Message "
		    + chordMessage.msgType + " to " + chordMessage.intNodeID);
	    e.printStackTrace();
	}

	Logger.log(LogType.MESSAGE_SENT, chordMessage.msgType + " to Node "
		+ getNodeID(dst));
    }
}